package rediscommand

import (
	"context"
	"dataflow-operator/command"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/go-logr/logr"
	"github.com/go-redis/redis/v8"
)

type RedisCommander struct {
	RedisClient *redis.Client
	Logger      logr.Logger
}

type DataflowAppNum struct {
	DFMNumber int `json:"dfmNumber"`
	ROSNumber int `json:"rosNumber"`
	EBNumber  int `json:"ebNumber"`
	PSNumber  int `json:"psNumber"`
	ESNumber  int `json:"esNumber"`
}

var (
	appNum DataflowAppNum
	//检测到Booted+Running的次数
	preStartNumber = 0
)

func (rc *RedisCommander) Start(ctx context.Context) {
	//初始化redis命令的channel
	RedisControlChannel = fmt.Sprintf(RedisAppCmdKeyPattern, Partition)
	for {
		select {
		case <-ctx.Done():
			return
		case conf := <-command.DataflowConfigChannel:
			jsonStr := conf.Data["apps"]
			err := json.Unmarshal([]byte(jsonStr), &appNum)
			if err != nil {
				rc.Logger.Error(err, "Unmarshal DataflowAppNum error")

			}
		default:
			appStateNames := getAppStateNames(appNum)
			appStatus, err := rc.getAppStatus(appStateNames)
			if err != nil {
				rc.Logger.Error(err, "getAppStatus error")
			} else {
				if rc.shouldStop(appStatus) {
					//App发生故障时发送stop
					rc.Logger.Info("Apps are breakdown, stop them.")
					rc.processCMD(command.StopCMD)
				} else if rc.shouldStart(appStatus) {
					//App恢复时发送start
					rc.Logger.Info("Apps are backing to normal, start them.")
					rc.processCMD(command.StartCMD)
				} else {
					rc.Logger.Info("Apps are running normally, do nothing.")
				}
			}
			time.Sleep(8 * time.Second)
		}
	}
}

func (rc *RedisCommander) getAppStatus(appNames []string) ([]interface{}, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if appNum.DFMNumber == 0 || appNum.EBNumber == 0 || appNum.ESNumber == 0 ||
		appNum.PSNumber == 0 || appNum.ROSNumber == 0 {
		// 配置文件不完整
		return nil, errors.New("App numbers are missing")
	}

	sc, err := rc.RedisClient.MGet(ctx, appNames...).Result()
	if err != nil {
		rc.Logger.Error(err, "Get app status from redis error.")
		return nil, errors.New("get app state error")
	}
	return sc, nil
}

func (rc *RedisCommander) shouldStart(appStatus []interface{}) bool {
	bootedNum := 0
	runningNum := 0

	//统计app状态
	for _, v := range appStatus {
		if v == nil {
			return false
		} else if v.(string) == "Booted" {
			bootedNum++
		} else if v.(string) == "Running" {
			runningNum++
		}
	}
	if bootedNum > 0 && runningNum > 0 {
		preStartNumber++
	}

	//所有app都为Booted则可以启动
	if bootedNum == len(appStatus) {
		preStartNumber = 0
		return true
	}

	//三次以上app处于Booted+Running可以启动
	if preStartNumber >= 3 {
		preStartNumber = 0
		return true
	}

	return false
}

func (rc *RedisCommander) shouldStop(appStatus []interface{}) bool {
	runningNum := 0
	noneNum := 0

	for _, v := range appStatus {
		if v == nil {
			noneNum++
			continue
		}
		if v.(string) == "Running" {
			runningNum++
		}
	}
	if runningNum > 0 && noneNum > 0 {
		return true
	}

	return false
}

func (rc *RedisCommander) processCMD(cmd string) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if cmd == command.StartCMD {
		// 发送start_timestamp
		startTimeKey := fmt.Sprintf("df:%s.start_timestamp", Partition)
		rc.RedisClient.Set(ctx, startTimeKey, time.Now().UnixNano(), 0)
	}

	res, err := rc.RedisClient.Publish(ctx, RedisControlChannel, cmd).Result()
	if err != nil {
		rc.Logger.Error(err, "Can not connect to redis")
		return
	}

	if res == 0 {
		rc.Logger.Error(err, "Send command failed, no app receive command")
	} else {
		rc.Logger.Info("Send command success")
	}
}

func getAppStateNames(appNum DataflowAppNum) []string {
	appNames := []string{}

	for i := 0; i < appNum.DFMNumber; i++ {
		appName := fmt.Sprintf(RedisAppStateKeyPattern, Partition, "dfm-"+strconv.Itoa(i))
		appNames = append(appNames, appName)
	}

	for i := 0; i < appNum.ROSNumber; i++ {
		appName := fmt.Sprintf(RedisAppStateKeyPattern, Partition, "ros-"+strconv.Itoa(i))
		appNames = append(appNames, appName)
	}

	for i := 0; i < appNum.EBNumber; i++ {
		appName := fmt.Sprintf(RedisAppStateKeyPattern, Partition, "builder-"+strconv.Itoa(i))
		appNames = append(appNames, appName)
	}

	for i := 0; i < appNum.ESNumber; i++ {
		appName := fmt.Sprintf(RedisAppStateKeyPattern, Partition, "es-"+strconv.Itoa(i))
		appNames = append(appNames, appName)
	}

	for i := 0; i < appNum.PSNumber; i++ {
		appName := fmt.Sprintf(RedisAppStateKeyPattern, Partition, "ps-"+strconv.Itoa(i))
		appNames = append(appNames, appName)
	}

	return appNames
}
